package cn.gwm.flink.streaming.task;

import cn.gwm.flink.streaming.beans.BeanSource;
import cn.gwm.flink.streaming.config.GlobalConfigSingleton;
import cn.gwm.flink.streaming.dwm.charge.model.ChargeStateEnum;
import cn.gwm.flink.streaming.dws.reservecharge.ReserveChargeData;
import cn.gwm.flink.streaming.dws.reservecharge.ReserveChargeRichFlatMapFunction;
import cn.gwm.flink.streaming.sink.kafka.FlinkKafkaUtil;
import cn.gwm.utils.ConfigLoader;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author Liu.Nanfang
 * @date 2022/6/30 13:24
 * @description 预约充电 DWS 层
 * 1. 获取根据 dwm 层清洗过的充电主题数据
 * 2. 根据数据进行逻辑处理
 * 2.1 查找电价信息，并计算设置
 * 2.2 计算充电时长
 * 2.3
 * 3. 根据 endChargeTime chargeStatus 判断是否充电结束 存入一个完整的充电过程至 hbase
 * 4. 根据实时充电状态数据，推送至业务kafka
 */
@Slf4j
public class DwsReserveChargeTask extends BaseTask {
    public static Logger logger = LoggerFactory.getLogger(BaseTask.class);

    private static final OutputTag<ReserveChargeData> reserveChargeDataHbase = new OutputTag<ReserveChargeData>("reserveChargeDataHbase") {
    };

    private static final String DASK_NAME = "dws-reserveCharge";


    public static void main(String[] args) throws Exception {
        try {
            ConfigLoader.init(args);
            StreamExecutionEnvironment env = getEnv(DASK_NAME);
            env.enableCheckpointing(120 * 1000L, CheckpointingMode.EXACTLY_ONCE);
            env.getCheckpointConfig().setCheckpointTimeout(240 * 1000L);
            env.setParallelism(6);
            execute(env);
            env.execute(DASK_NAME);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static void execute(StreamExecutionEnvironment env) throws Exception {
        final FlinkKafkaConsumer<String> kafkaConsumer = FlinkKafkaUtil.getConsumerGroupOffsets(
                "consumer-dws-receive",
                ConfigLoader.get("kafka.topic.charge.summar")
        );
//        kafkaConsumer.setStartFromTimestamp(1678512890383L);
        GlobalConfigSingleton grobalConfig = GlobalConfigSingleton.getInstance();
        DataStream<ReserveChargeData> sourceDataStream = env.addSource(kafkaConsumer)
                .map(str -> {
                    if (JSONUtil.isTypeJSON(str)) {
                        return JSONUtil.toBean(str, ReserveChargeData.class);
                    }
                    log.error("Illegal JSON message: {}", str);
                    return null;
                })
                .name("kafka source and Parse JSON to object").filter(item -> item != null);
        // 取消异步调用
//        SingleOutputStreamOperator<ReserveChargeData> processGps = AsyncDataStream.unorderedWait(sourceDataStream, new HttpAsyncGpsFunction(),
//                0, TimeUnit.MILLISECONDS, 20).name("handle gps");


        SingleOutputStreamOperator<ReserveChargeData> process = sourceDataStream
//                .map(new ReserveChargeGpsMapFunction())
                .keyBy(sd -> sd.getDeviceId())
                .flatMap(new ReserveChargeRichFlatMapFunction())
                .process(new SelectFinishData())
                .name("handle data and split data stream").disableChaining();

//        ReserveChargeDwsTaskTest.localTestProcessSaveFile(process);
        defaultSinkToKafka(process, ConfigLoader.get("kafka.dws_receive.topic"), "id", null, null);
        process
                .filter(item -> ObjectUtil.isNotNull(item.getEndFlag()))
                .map(item -> (JSONObject) JSONUtil.parseObj(item))
                .addSink(getFileSink(ConfigLoader.get("hive.table.name.dws.charge.reserve"),
                        ".txt",
                        "dws/" + ConfigLoader.get("hive.table.name.dws.charge.reserve"),
                        BeanSource.SourceEnum.chargeReserve))
                .name("addSink hive");
        process.print();


    }


    private static final class SelectFinishData extends ProcessFunction<ReserveChargeData, ReserveChargeData> {
        private static final long serialVersionUID = 1L;

        @Override
        public void processElement(ReserveChargeData value, Context context, Collector<ReserveChargeData> out) throws Exception {
            // 完成的数据存入hbase 进行更新操作
            if (ChargeStateEnum.END.getCode().equals(value.getChargeStatus()) && value.getEnd_chrgConnectTid() != null) {
                context.output(reserveChargeDataHbase, value);
            }
            out.collect(value);
        }
    }


}
